package preprocess;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import utils.Utils;


public class DumpUniqueAuctionItem extends Configured implements Tool {
	enum Counter{INVALID_PV_CLICK_SEARCH, AUCTION_53, AUCTION_34,
		INVALID_AUCTION, ITEM_REPLICED, AUCTION_NOT_FOUND, ITEM_JOINED};
	public static String getUniqueAuctionItemPath(String date) {
		return Utils.rootDir + date + "/UniqueAuctionItem/";
	}
	public static class PvClickSearchMap extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			// TODO Auto-generated method stub
			String[] values = value.toString().split(Utils.separator);
			if (values.length != 30) {
				reporter.incrCounter(Counter.INVALID_PV_CLICK_SEARCH, 1);
				return;
			}
			String[] clicks = values[29].split("\u0002");
			HashMap<String, Integer> clickItem = new HashMap<String, Integer>();
			for (String click : clicks) {
				if (Utils.isValid(click)) {
					String[] clickInfo = click.split("\u0001");
					if (clickInfo.length == 5 && Utils.isValid(clickInfo[3])) {
						String item = clickInfo[3];
						if (clickItem.containsKey(item))
							clickItem.put(item, clickItem.get(item) + 1);
						else 
							clickItem.put(item, 1);
					}
				}
				
			}
			String[] itemTypes = values[28].split(Utils.pvAuctionSeparator1);
			for (String itemStr : itemTypes) {
				String[] itemList = itemStr.split(Utils.pvAuctionSeparator2);
				for (String item : itemList) {
					k.set(item);
					int pvCount = 1;
					int clickCount = 0;
					if (clickItem.containsKey(item))
						clickCount = clickItem.get(item);
					v.set(Utils.getRecord(new String[]{String.valueOf(pvCount), String.valueOf(clickCount)}, Utils.separator));
					output.collect(k, v);
				}
			}
		}
		
	}
	
	public static class AuctionMap extends MapReduceBase implements Mapper<BytesWritable, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		public void map(BytesWritable key, Text value, 
				OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			String[] values = value.toString().split(Utils.auctionSeparator);
			if (values.length == 53) {
				reporter.incrCounter(Counter.AUCTION_53, 1);
				
			} else if (values.length == 34) {
				reporter.incrCounter(Counter.AUCTION_34, 1);
			} else if (values.length < 34){
				reporter.incrCounter(Counter.INVALID_AUCTION, 1);
				return;
			}
			
			k.set(values[18]);
			v.set(Utils.getRecord(values, Utils.separator, 0, 34));
			output.collect(k, v);
		}
	}
	
	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		@Override
		public void reduce(Text key, Iterator<Text> iter, OutputCollector<Text, Text> output, 
				Reporter reporter) throws IOException {
			int pvCount = 0;
			int clickCount = 0;
			String[] itemColumns = null;
			while (iter.hasNext()) {
				String[] values = iter.next().toString().split(Utils.separator);
				if (values.length == 2) {
					pvCount += Integer.parseInt(values[0]);
					clickCount += Integer.parseInt(values[1]);					
				}
				if (values.length == 34) {
					if (itemColumns != null) {
						reporter.incrCounter(Counter.ITEM_REPLICED, 1);
						continue;
					}
					itemColumns = values;
				}
			}
			if (pvCount > 0 && itemColumns != null) {
				
				k.set(itemColumns[0]);
				v.set(Utils.getRecord(new String[]{itemColumns[18], itemColumns[1], 
						itemColumns[30], String.valueOf(pvCount), 
						String.valueOf(clickCount), 
						itemColumns[9]}, 
						Utils.separator));
				output.collect(k, v);
				reporter.incrCounter(Counter.ITEM_JOINED, 1);
			}
			if (itemColumns == null) {
				reporter.incrCounter(Counter.AUCTION_NOT_FOUND, 1);
			}
		}
	}
	
	/*
	 * (non-Javadoc)
	 * DumpUniqueAuctionItemed Table
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/DumpUniqueAuctionItemed/
	 * 1.item id 32位
	 * 2.item id 数字
	 * 3.卖家32位id
	 * 4.卖家数字id
	 * 5.pv count
	 * --------------新增-------------------
	 * 6.click count
	 * 7.商品类目id
	 */
	@Override
	public int run(String[] args) throws Exception {
		// input
		String pvSearchClickFile = SelectPvSearchClickByBuyerSamples.getSelectPvSearchClickByBuyerSamplesedPath(args[0]);
		String itemFile = "/group/taobao/taobao/hive/r_auction_auctions/pt=" + args[0] + "000000/if_online=1";
		
		// Output
		String output = getUniqueAuctionItemPath(args[0]);
		System.out.println("start job DumpUniqueAcutionItem...");
		System.out.println("pvSearchClickFile: " + pvSearchClickFile);
		System.out.println("itemFile: " + itemFile);

		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		JobConf conf = new JobConf(getConf(), PvSearchClickJoiner.class);
		
		conf.setJobName("DumpUniqueAcutionItem");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		MultipleInputs.addInputPath(conf, 
				new Path(pvSearchClickFile), 
				TextInputFormat.class, 
				PvClickSearchMap.class);
		
	
		MultipleInputs.addInputPath(conf, 
				new Path(itemFile),
				SequenceFileInputFormat.class, 
				AuctionMap.class);
		
		FileOutputFormat.setOutputPath(conf, new Path(output));
		conf.setReducerClass(Reduce.class);
		JobClient.runJob(conf);
		return 0;
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new DumpUniqueAuctionItem(), args);
		System.exit(res);
	}
}
